package com.ayu.flink.timelineandwindow;

import com.ayu.flink.datasteamapibase.ClickSource;
import com.ayu.flink.datasteamapibase.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @Author: 徐林玉//作者及
 * @Date: 2022/3/23//完成日期
 * @Description: //
 * @Version: v0.0.1 // 版本信息
 * @Function List: // 主要函数及其功能
 * @Others: // 其它内容的说明
 * @History: // 历史修改记录
 **/
public class UvCountExample
{
	public static void main(String[] args) throws Exception
	{


		//1、创建执行环境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);


		//2、获取数据源
		DataStreamSource<Event> eventDataStreamSource = env.addSource(new ClickSource());

		//3、设置水位线和时间戳
		SingleOutputStreamOperator<Event> stream = eventDataStreamSource.assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps().withTimestampAssigner((SerializableTimestampAssigner<Event>) (element, recordTimestamp) -> element.timestamp));

		//4、需要按照url分组，开滚动窗口统计 使用aggregate函数
		stream.keyBy(data->data.url)
				.window(TumblingEventTimeWindows.of(Time.seconds(10)))
				.aggregate(new UrlCountAgg(),new UrlCountResult())
				.print();

		//7、执行
		env.execute();

	}


	public static class UrlCountAgg implements AggregateFunction<Event,Long,Long>{

		@Override
		public Long createAccumulator()
		{
			return 0L;
		}

		@Override
		public Long add(Event event, Long accumulator)
		{
			return accumulator+1;
		}

		@Override
		public Long getResult(Long accumulator)
		{
			return accumulator;
		}

		@Override
		public Long merge(Long aLong, Long acc1)
		{
			return null;
		}
	}


	public static class UrlCountResult extends ProcessWindowFunction<Long,UrlViewCountPoJo,String, TimeWindow>
	{

		@Override
		public void process(String url, Context context, Iterable<Long> elements, Collector<UrlViewCountPoJo> out) throws Exception
		{
			// 结合窗口信息，包装输出内容
			Long start = context.window().getStart();
			Long end = context.window().getEnd();
			// 迭代器中只有一个元素，就是增量聚合函数的计算结果
			out.collect(new UrlViewCountPoJo(url, elements.iterator().next(), start, end));

		}
	}

}

